package com.spider.dao.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import javax.inject.Qualifier;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseDAOImp implements HBaseDao {
	HConnection hTablePool = null;
	Configuration conf = null;

	public HBaseDAOImp() {
		conf = new Configuration();
		conf = HBaseConfiguration.create();
		conf.set("hbase.rootdir", "hdfs://master1:9000/hbase");
		conf.set("hbase.zookeeper.quorum", "slave1,slave2");
		try {
			hTablePool = HConnectionManager.createConnection(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void create(String tableName, String[] cfs) {
		HBaseAdmin admin = null;
		try {
			admin = new HBaseAdmin(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
		try {
			if (admin.tableExists(tableName)) {
				System.out.println("���Ѵ��ڡ�");
			} else {
				HTableDescriptor tableDesc = new HTableDescriptor(tableName);
				for (int i = 0; i < cfs.length; i++) {
					//���ú��дز��ܶ�̬���ӣ������ǿ��Զ�̬���ӵģ�����hbase�����Ե�һ�����֡�
					tableDesc.addFamily(new HColumnDescriptor(cfs[i]));
				}
				admin.createTable(tableDesc);
				System.out.println("�����ɹ���");
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	@Override
	public void save(Put put, String tableName) {
		HTableInterface table = null;

		try {
			table = hTablePool.getTable(tableName);
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public void save(List<Put> put, String tableName) {
		HTableInterface table = null;

		try {
			table = hTablePool.getTable(tableName);
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public void insert(String tableName, String rowKey, String family, String quailifer, String value) {
		HTableInterface table = null;

		try {
			table = hTablePool.getTable(tableName);
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(family), Bytes.toBytes(quailifer), Bytes.toBytes(value));
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public void insert(String tableName, String rowKey, String family, String[] quailifer, String[] value) {
		HTableInterface table = null;

		try {
			table = hTablePool.getTable(tableName);
			Put put = new Put(Bytes.toBytes(rowKey));
			for (int i = 0; i < quailifer.length; i++) {
				put.add(Bytes.toBytes(family), Bytes.toBytes(quailifer[i]), Bytes.toBytes(value[i]));
				put.setWriteToWAL(false);// ����дWAL��־���������д�������
			}
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public Result getRow(String tableName, String rowKey) {
		HTableInterface table = null;

		Result rsResult = null;
		try {
			table = hTablePool.getTable(tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			rsResult = table.get(get);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return rsResult;
	}

	@Override
	public List<Result> getRows(String tableName, String rowKeyLike) {
		HTableInterface table = null;

		List<Result> list = null;
		try {
			table = hTablePool.getTable(tableName);
			PrefixFilter filter = new PrefixFilter(Bytes.toBytes(rowKeyLike));
			Scan scan = new Scan();
			scan.setFilter(filter);
			ResultScanner scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}


	@Override
	public List<Result> getRows(String tableName, String startRow, String stopRow) {
		HTableInterface table = null;
		List<Result> list = null;
		try {
			table = hTablePool.getTable(tableName);
			Scan scan = new Scan();
			scan.setStartRow(Bytes.toBytes(startRow));
			scan.setStopRow(Bytes.toBytes(stopRow));
			ResultScanner scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rsResult : scanner) {
				list.add(rsResult);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}

	@Override
	public void deleteRecords(String tableName, String rowKeyLike) {
		HTableInterface table = null;
		try {
			table = hTablePool.getTable(tableName);
			PrefixFilter filter = new PrefixFilter(Bytes.toBytes(rowKeyLike));
			Scan scan = new Scan();
			scan.setFilter(filter);
			ResultScanner scanner = table.getScanner(scan);
			List<Delete> list = new ArrayList<Delete>();
			for (Result rs : scanner) {
				Delete del = new Delete(rs.getRow());
				list.add(del);
			}
			table.delete(list);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
}
